package tsif.tcluster.master;

import java.lang.reflect.Method;
import java.util.Arrays;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicInteger;

import tsif.tcluster.TClusterMaster;
import tsif.tcluster.service.proxy.TClusterServiceProxy;
import tsif.tcluster.service.proxy.TClusterServices;

/**
 * 本地中央控制<br>
 * 
 */
public class TChlusterLocalMaster extends TClusterMaster {
	protected ConcurrentMap<Integer, ContextNode> contexts;
	protected ConcurrentMap<String, Long> uniqueServiceId;
	protected AtomicInteger idCounter;

	@Override
	public boolean init(TClusterServiceProxy serviceProxy, long serviceId, Object... args) throws Exception {
		if (!super.init(serviceProxy, serviceId, args)) {
			return false;
		}
		contexts = new ConcurrentHashMap<>();
		uniqueServiceId = new ConcurrentHashMap<>();
		idCounter = new AtomicInteger();
		return true;
	}

	/** 注册集群节点, 如果已存在, 检测是否冲突. **/
	protected boolean registerContext(ContextNode node) {
		// try put
		ContextNode old = contexts.putIfAbsent(node.getId(), node);
		if (old == null) {
			return true; // put success
		}
		// check same
		if (!old.getHost().equals(node.getHost()) || old.getPort() != node.getPort()) {
			return false;
		}
		// is same, update time
		old.setUpdateTime(System.currentTimeMillis());
		return true;
	}

	@Override
	public int registerContext(int contextId, String host, int port) {
		// create context
		ContextNode node = new ContextNode();
		node.setHost(host);
		node.setPort(port);
		node.setUpdateTime(System.currentTimeMillis());

		// set contextId
		if (contextId != 0) {
			node.setId(contextId); // 指定集群Id
		} else {
			node.setId(newContextId());
		}

		// register context
		if (!registerContext(node)) {
			throw new RuntimeException("register context fail! contextId=" + contextId + " host=" + host + " port=" + port);
		}
		return node.getId();
	}

	/** 获取一个集群Id(全局唯一) **/
	protected int newContextId() {
		for (int i = 0; i < 100; i++) {
			int contextId = idCounter.incrementAndGet();
			if (!contexts.containsKey(contextId)) {
				return contextId;
			}
		}
		throw new RuntimeException("can't get new context id!");
	}

	@Override
	public ContextNode getContext(int contextId) {
		return contexts.get(contextId);
	}

	@Override
	public boolean removeContext(int contextId) {
		ContextNode contextNode = contexts.remove(contextId);
		if (contextNode == null) {
			return false;
		}
		System.out.println("context remove: " + contextNode);
		return true;
	}

	@Override
	public void updateContext(int contextId) {
		ContextNode contextNode = contexts.get(contextId);
		if (contextNode == null) {
			return;
		}
		contextNode.setUpdateTime(System.currentTimeMillis());
	}

	@Override
	public long bindUniqueService(String key, long serviceId, boolean replace) {
		// serviceId 小于0 并替换为删除.
		if (serviceId <= 0) {
			throw new RuntimeException("can't bind service by empty!");
		}

		// replace
		if (replace) {
			uniqueServiceId.put(key, serviceId);
			return serviceId;
		}

		// set serviceId
		Long old = uniqueServiceId.putIfAbsent(key, serviceId);
		if (old != null) {
			return old;
		}
		return serviceId;
	}

	@Override
	public void unbindUniqueService(String key, long serviceId) {
		uniqueServiceId.remove(key, serviceId);
	}

	@Override
	public long getUniqueService(String key) {
		Long id = uniqueServiceId.get(key);
		return (id != null) ? id : 0L;
	}

	@Override
	public void abort(int shutdownTime, boolean shutdownSelf) {
		// self context
		int selfContextId = context.getContextId();

		// abort other
		for (ContextNode contextNode : contexts.values()) {
			int contextId = contextNode.getId();
			if (selfContextId == contextId) {
				// continue; // self is last
			}

			try {
				long serviceId = TClusterServices.getLocalServiceId(contextId);
				this.accessor.call(serviceId, "shutdown", shutdownTime);
			} catch (Exception e) {
				System.err.println("abort context error! contextId=" + contextId + " selfContextId=" + selfContextId);
				e.printStackTrace();
			}
		}
		// // contexts.clear();
		// if (shutdownSelf) {
		// // shutdown self
		// try {
		// long serviceId = TClusterServices.getLocalServiceId(selfContextId);
		// this.accessor.send(serviceId, "shutdown", shutdownTime); // will timeout!
		// } catch (Exception e) {
		// System.err.println("abort context error! contextId=" + selfContextId);
		// e.printStackTrace();
		// }
		// }
	}

	@Override
	public void destroy() {
	}

	@Override
	public void update() {
		updateContexts(10 * 1000);
	}

	/** 更新集群信息, 如果某个集群环境很久没更新, 当做断开. **/
	protected void updateContexts(int timeout) {
		// 不执行超时移除, 因为有些注册了, 初始化很久导致移除无法恢复, 再者避免已经断线, 其他人也连不上.

		// long nowTime = System.currentTimeMillis();
		//
		// // check timeout
		// List<ContextNode> removes = new ArrayList<>();
		// for (ContextNode node : contexts.values()) {
		// long dt = nowTime - node.getUpdateTime();
		// if (dt < timeout) {
		// continue;
		// }
		// removes.add(node);
		// }
		//
		// // check remove
		// if (removes.isEmpty()) {
		// return;
		// }
		//
		// // remove
		// for (ContextNode node : removes) {
		// // remove
		// if (!contexts.remove(node.getId(), node)) {
		// continue;
		// }
		// // show
		// System.out.println("context timeout remove: " + node);
		// }

	}

	@Override
	public void dispose(long serviceId, String funcName, Object[] args, IReturn<Object> ret) {
		Method method = getMethodByName(TChlusterLocalMaster.class, funcName);
		if (method == null) {
			throw new RuntimeException("no find method by name! " + funcName);
		}
		// 执行
		try {
			Object retObj = method.invoke(this, args);
			ret.ret(1, null, retObj);
		} catch (Exception e) {
			System.err.println("dispose error! (TChlusterLocalMaster.java:98) funcName=" + funcName + " args=" + Arrays.toString(args));
			throw new RuntimeException(e);
		}

	}

	/** 根据函数名获取函数 **/
	private static Method getMethodByName(Class<?> clazz, String funcName) {
		for (Method method : clazz.getMethods()) {
			if (method.getName().equals(funcName)) {
				return method;
			}
		}
		return null;
	}
}
